/*
 * 文件名： ReportDataTask.java
 *
 * 创建日期： 2016年12月5日
 *
 * Copyright(C) 2016, by <a href="mailto:liws@xingyuanauto.com">liws</a>.
 *
 * 原始作者: liws
 *
 */
package com.moli.iov.gateway.server.task;

import static com.moli.iov.gateway.util.ResponseBytesUtils.joinRespBytes;
import static com.moli.iov.protocol.enums.CommandTag.REALTIME_INFO_REPORT;
import static com.moli.iov.protocol.enums.CommandTag.TERMINAL_TIME_CHECK;

import com.moli.iov.base.remote.IVehicleTboxRemoteService;
import com.moli.iov.gateway.server.NettyChannelManager;
import com.moli.iov.gateway.server.VinThreadPoolExecutor;
import com.moli.iov.gateway.server.handle.CommandInvoker;
import com.moli.iov.gateway.server.model.KafkaSendData;
import com.moli.iov.gateway.server.model.UpStreamData;
import com.moli.iov.gateway.util.GatewayUtil;
import com.moli.iov.gateway.util.ResponseBytesUtils;
import com.moli.iov.gateway.util.VinRouter;
import com.moli.iov.gateway.util.logger.GatewayLogger;
import com.moli.iov.protocol.codec.DataWrapper;
import com.moli.iov.protocol.codec.DownStreamData;
import com.moli.iov.protocol.codec.ReqStreamData;
import com.moli.iov.protocol.enums.CommandTag;
import com.moli.iov.protocol.enums.DateTimeTag;
import com.moli.iov.protocol.enums.TopicEnums;
import com.moli.iov.protocol.util.BccUtils;
import com.moli.iov.util.ByteArrayUtil;
import com.moli.iov.util.PropertiesUtils;
import com.moli.iov.util.SpringContextUtils;
import com.moli.iov.util.StringUtil;
import io.netty.channel.Channel;
import java.util.Date;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * 数据上报 多线程任务Task
 *
 * @author <a href="mailto:liws@xingyuanauto.com">liws</a>
 * @version $Revision$
 * @since 2016年12月5日
 */
public class UpStreamDataTask implements Runnable {

  private static final Logger logger = LoggerFactory.getLogger(UpStreamDataTask.class);

  private IVehicleTboxRemoteService iVehicleTboxService;


  private VinRouter vinRouter;

  /**
   * 下行线程池
   */
  private VinThreadPoolExecutor downThreadPool;

  /**
   * kafka数据发送线程池
   */
  private VinThreadPoolExecutor kafkaSendThreadPool;
  /**
   * 上行报文封装对象
   */
  private UpStreamData usd;

  private String vinLogPath;


  /**
   * @author liws
   */
  public UpStreamDataTask(UpStreamData usd) {
    this.usd = usd;
    downThreadPool = SpringContextUtils.getBean("downThreadPool");
    kafkaSendThreadPool = SpringContextUtils.getBean("kafkaSendThreadPool");
    iVehicleTboxService = SpringContextUtils.getBean("iVehicleTboxRemoteService");
    vinLogPath = PropertiesUtils.getProperty("vin.log.path");
    vinRouter = SpringContextUtils.getBean("vinRouter");
  }


  @Override
  public void run() {

    try {
      if (usd == null || usd.channel == null) {
        logger.error("usd有误");
        return;
      }
      byte[] bytes = usd.dataBytes;
      if (bytes == null || bytes.length < 3) {
        logger.error("原始报文误：" + ByteArrayUtil.toHexString(bytes));
      }
      String vin = GatewayUtil.getVin(bytes);
      if (StringUtil.isBlank(vin) || vin.trim().length() != 17) {
        logger.error("[上行报文][VIN格式不对][Vin=" + vin + "]");
        return;
      }
      boolean checkBccCode = BccUtils.checkBccCode(bytes);
      if (!checkBccCode) {
        logger.error(
            "[上行报文 bbc亦或校验有误][Vin=" + vin + "][" + ByteArrayUtil.toHexString(usd.dataBytes) + "]");
        return;
      }
      CommandTag command = CommandTag.valOf(bytes[2]);
      logger.info("[上行报文:" + command.getK() + "][vin=" + vin + "][" + ByteArrayUtil
          .toHexString(usd.dataBytes) + "]");
      GatewayLogger.getLogger(vin, vinLogPath).info("[RECV]:" + ByteArrayUtil.toHexString(bytes));
      String packetTopoc = TopicEnums.getTopic(command.getV());
      if (StringUtil.isBlank(packetTopoc)) {
        logger.error("上行topic为空,bytes:{}", ByteArrayUtil.toHexString(bytes));
        return;
      }
      //下行命令终端响应 报文
      if (CommandTag.REMOTE_CONTROL_RESPONSE == command
          || CommandTag.REMOTE_CONTROL_RESULT_RESPONSE == command
          || CommandTag.REMOTE_UPGRADE_RESPONSE == command
          || CommandTag.TERMINAL_PARAMETER_QUERY_RESPONSE == command
          || CommandTag.TERMINAL_PARAMETER_SETTING_RESPONSE == command
          || CommandTag.TERMINAL_CONTROL_RESPONSE == command
          || CommandTag.REMOTE_UPGRADE_RECV_RESPONSE == command) {
        //下行命令 响应报文处理
        downResponseHandle(packetTopoc, vin, bytes);
        return;
      }
      //上行报文，命令报文
      if ((bytes[3] + 256) == 0xfe) {
        switch (command) {
          case VEHICLE_REGISTER:  //处理注册报文
            if (vin != null && !vin.equals("LNBSCB3F4HW162527")) {
              handleRegister(packetTopoc, vin);
            }
            break;
          case VEHICLE_LOGIN:  //处理登陆报文
            CommandInvoker.invoke(CommandTag.VEHICLE_LOGIN, usd);
            break;
          case TERMINAL_TIME_CHECK://校时
          case REALTIME_INFO_REPORT://实时
          case RESEND_INFO_REPORT://补发
          case VEHICLE_LOOUT://车辆登出
          case HEARTBEAT://心跳
            // TODO: 2019/1/16 方便测试临时加入,上线需删除
            /**************方便压测临时代码***************/
//                        pressureTest(vin);
            /**************方便压测临时代码***************/
            //处理终端校时、心跳、实时、补发、登出
            handle(command, packetTopoc, vin, bytes);
            break;
          default:
            //不处理，诸如23 23 01 01 等bcc亦或校验同时长度都正确但却不属于服务端要求回复的也不是主动上报的
            logger.error("[上行报文 非法报文][Vin=" + vin + "][" + ByteArrayUtil.toHexString(bytes) + "]");
            break;
        }

      }
    } catch (Exception e) {
      logger.error("数据上报 多线程任务Task 处理异常", e);
    }

  }

  /**
   * 临时压力测试使用，不校验登陆
   */
  private void pressureTest(String vin) {
    Channel channelByUid = NettyChannelManager.getInstance().getLoginChannelByVin(vin);
    if (channelByUid == null) {
      Channel channel = NettyChannelManager.getInstance().getLoginChannelByVin(vin);
      //同一个VIN不同的channel登陆，后者挤掉前者
      if (channel != null && !channel.id().asLongText().equals(usd.channel.id().asLongText())) {
        channel.close();
//                NettyChannelManager.getInstance().removeChannel(channel);
      }
      //注册新的通道han
      String channelId = usd.channel.id().asLongText().trim();
      NettyChannelManager.getInstance().bindVinAndChannel(vin, usd.channel);//绑定uid与通道ID新关系
      //移除注册通道
      Channel registerChannel = NettyChannelManager.getInstance().getRegisterChannelByVin(vin);
      if (registerChannel != null && registerChannel.id().asLongText()
          .equals(usd.channel.id().asLongText())) {
        NettyChannelManager.getInstance().removeRegisterChannel(registerChannel);
      }

    }
  }

  /**
   * 下行命令响应报文
   */
  public void downResponseHandle(String topic, String vin, byte[] bytes) {
    logger.info(" [其他上行命令报文] [下行命令响应报文] topic: {}， vin: {}, bytes: {}", topic, vin, bytes);
    DataWrapper dataWrapper = new DataWrapper("", bytes, new Date());
    this.sendMsg(topic, vin, dataWrapper);
  }

  /**
   * 处理上行的终端校时，实时，补发，车辆登出逻辑
   *
   * @param command 上行命令
   * @param topic kafka主题
   * @param vin 车辆vin
   * @param bytes 上行的原始报文
   */
  private void handle(CommandTag command, String topic, String vin, byte[] bytes) {
    //登陆成功后缓存的channel,后续报文一律使用登陆后创建的chanel
    Channel loginChannel = NettyChannelManager.getInstance().getLoginChannelByVin(vin);
    //正在通信使用的channel
    Channel channel = usd.channel;
    if (loginChannel == null) {
      logger.error("车辆未登陆,vin:{}，bytes:{}", vin, ByteArrayUtil.toHexString(usd.dataBytes));
      channel.close();
      return;
    }
    //未使用登陆后的channel上发报文
    if (!loginChannel.id().asLongText().equals(channel.id().asLongText())) {
      logger.error("通道无效，未使用登陆后的channel上发报文，vinChannel={},vin={}", loginChannel, vin);
      channel.close();
      return;
    }
    DownStreamData dsd = null;
    if (command == TERMINAL_TIME_CHECK) {//添加时间以便终端校时
      dsd = new DownStreamData(vin,
          ResponseBytesUtils.joinRespBytes_resetTime(bytes, new DateTimeTag().toBytes(), true));
    } else {
      dsd = new DownStreamData(vin, joinRespBytes(bytes, true));
    }
    //数据下行至终端
    if (dsd != null) {
      downThreadPool.execute(vin, new DownStreamDataTask(dsd));
    }
    //只有实时和补发业务数据才发送kafka，校时，登出，心跳只回业务不处理
    if (command == REALTIME_INFO_REPORT || command == REALTIME_INFO_REPORT) {
      this.sendMsg(topic, vin, new ReqStreamData(usd.reqDateTime, usd.dataBytes));
    }
  }

  /**
   * 处理注册逻辑
   */
  private void handleRegister(String topic, String vin) {
    Channel connectingChannel = usd.channel;
    try {
      // 不同的信道都发送同样VIN的注册数据的时候，后来在挤掉前面的信道
      if (StringUtil.isNotBlank(vin)) {
        Channel oldChannel = NettyChannelManager.getInstance().getRegisterChannelByVin(vin);
        if (oldChannel != null && !oldChannel.id().asLongText()
            .equals(connectingChannel.id().asLongText())) {
//                    NettyChannelManager.getInstance().removeChannel(oldChannel);
          oldChannel.close();
          logger.error("[上行报文][不同的连接重复注册VIN，旧的连接将被断开！][Vin=" + vin + "]");
        }
      }
      //根据channelId查询已经登陆的车辆，已经登陆的发送注册报文不处理
      String activeVin = NettyChannelManager.getInstance()
          .getLonginVinByChannel(connectingChannel.id().asLongText().trim());
      if (StringUtil.isNotBlank(activeVin) && vin.equals(activeVin)) {
        logger.error("[上行报文][VIN已经登陆，重复注册][Vin=" + vin + "]");
        return;
      }
      //通道已经有其他车辆登陆，不允许此VIN注册
      if (StringUtil.isNotBlank(activeVin) && !vin.equals(activeVin)) {
        logger.error("[上行报文][该通道已经有其他车辆登陆，不允许此VIN注册][Vin=" + vin + "]");
        return;
      }
      if (StringUtil.isBlank(activeVin)) {
        NettyChannelManager.getInstance().bindUidAndRegisterChannel(vin, connectingChannel);
        this.sendMsg(topic, vin, new ReqStreamData(usd.reqDateTime, usd.dataBytes));
        logger.info("[上行报文][车辆注册][Vin=" + vin + "]");
      }
    } catch (Exception e) {
      e.printStackTrace();
      logger.error("[上行报文:车辆注册][Vin=" + vin + "][注册失败]");
    }
  }

  private void sendMsg(String topic, String vin, Object object) {
    try {
      KafkaSendData kafkaSendData = new KafkaSendData(topic, vin, object);
      kafkaSendThreadPool.execute(vin, new KafkaSendTask(kafkaSendData));
    } catch (Exception e) {
      logger.error("kafka发送异常", e);
    }
  }
}
